package com.sf.iisp.kafka.util;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import com.sf.kafka.api.produce.IKafkaProducer;
import com.sf.kafka.api.produce.ProduceConfig;
import com.sf.kafka.api.produce.ProducerPool;
import com.sf.kafka.exception.KafkaException;

/**
 * Created by Administrator on 2015/4/28.
 */
public class ProduceSendMsgUtil {
    public static void main(String[] args) throws KafkaException {
        // 生产者连接池大小
        final int poolSize = 1;
        // 主题名称
        String topic = "TEST_TOPIC_089245";
        // KAFKA连接地址
        String url = "http://10.0.137.112:8292/mom-mon/monitor/requestService.pub";
        // 主题所在的集群名称
        String clusterName = "other";
        // 主题名称＋分隔符 （固定不变）＋主题的校验码
        String topicTokens = topic + ":" + "12345678";

        // 简易模式
        ProduceConfig produceConfig = new ProduceConfig(poolSize, url, clusterName, topicTokens);
        IKafkaProducer kafkaProducer = new ProducerPool(produceConfig);

//        // 个性化参数模式
//        ProduceOptionalConfig optionalConfig = new ProduceOptionalConfig();
//        optionalConfig.setRequestRequiredAck(ProduceOptionalConfig.RequestRequiredAck.LEADER_REPLICA);
//        IKafkaProducer kafkaProducer = new ProducerPool(produceConfig, optionalConfig);

        send(kafkaProducer, topic);
        kafkaProducer.close();
    }

    private static void send(IKafkaProducer kafkaProducer, String topic) {
        SimpleDateFormat dateFormat = new SimpleDateFormat("MM-dd HH:mm:ss ");
        Date date = new Date();
        int count = 0;
        String time = dateFormat.format(date);

        // send bytes
//        count = sendSingleByte(kafkaProducer, topic, count, time);
//
//        // batch send bytes
//        count = batchSendBytes(kafkaProducer, topic, count, time);
//
//        // send string
//        count = sendSingleString(kafkaProducer, topic, count, time);

        // batch send string
        count = batchSendString(kafkaProducer, topic, count, time);

        System.out.println("send message count: " + count);
    }

    private static int batchSendString(IKafkaProducer kafkaProducer, String topic, int count, String time) {
        String message;
        List<String> messages = new ArrayList<String>();
        for (int i = 0; i < 5; i++) {
            message = time + ++count;
            messages.add(message);
        }

        for (int i = 0; i < 1 * 1; i++) {
            kafkaProducer.batchSendString(topic, messages);
        }
        return count;
    }

    private static int sendSingleString(IKafkaProducer kafkaProducer, String topic, int count, String time) {
        String message;
        message = time + ++count;
        kafkaProducer.sendString(topic, message);
        return count;
    }

    private static int batchSendBytes(IKafkaProducer kafkaProducer, String topic, int count, String time) {
        String message;
        List<byte[]> byteMessages = new ArrayList<byte[]>();
        for (int i = 0; i < 6; i++) {
            message = time + ++count;
            byteMessages.add(message.getBytes());
        }
        kafkaProducer.batchSendBytes(topic, byteMessages);
        return count;
    }

    private static int sendSingleByte(IKafkaProducer kafkaProducer, String topic, int count, String time) {
        String message;
        message = time + ++count;
        kafkaProducer.sendBytes(topic, message.getBytes());
        return count;
    }
}
